package nl.justobjects.pushlet.client;

import java.util.HashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import nl.justobjects.pushlet.core.Event;
import nl.justobjects.pushlet.core.Protocol;

import org.apache.log4j.Logger;

public class PushletClientListenerContainer implements PushletClientListener {
	
	private Logger Log = Logger.getLogger(PushletClientListenerContainer.class);

	private HashMap<String, PushletClientCallBack> subjectCallback; // 主题回调映射表
	private ExecutorService exec;
	private boolean asyn = false; // 是否异步执行
	private long hbTimestamp;
	
	public PushletClientListenerContainer(int nThreads, boolean asyn) {
		subjectCallback = new HashMap<String, PushletClientCallBack>();
		this.asyn = asyn;
		if (asyn)
			exec = Executors.newFixedThreadPool(nThreads);
	}

	public void stop() {
		if (asyn)
			exec.shutdownNow();
		subjectCallback.clear();
	}
	
	public synchronized void register(String subject, PushletClientCallBack callBack) {
		subjectCallback.put(subject, callBack);
	}
	
	public synchronized void unregister(String subject) {
		subjectCallback.remove(subject);
	}
	
	@Override
	public void onAbort(Event theEvent) {
		Log.debug("Remote Pushlet Server Abort.");
	}

	@Override
	public void onData(Event theEvent) {
		String subject = theEvent.getSubject();
		if (subject != null) {
			PushletClientCallBack callBack;
			synchronized (this) {
				callBack = subjectCallback.get(subject);
			}
			if (callBack != null) {
				if (asyn) {
                    callBack.setEvent(theEvent);
					exec.submit(callBack);
                } else {
					callBack.onEvent(theEvent);
                }
			}
		}
	}

	@Override
	public void onHeartbeat(Event theEvent) {
		Long pTime = (Long) theEvent.getField(Protocol.P_TIME);
		this.hbTimestamp = (pTime == null) ? -1l : pTime.longValue();
	}

	@Override
	public void onError(String message) {
		Log.error("[PushletClientListenerContainer]" + message);
	}

}
